package mapReduce.demo05_shuffle.demo06_groupGenderCount;

import mapReduce.util.JobSubmit;
import mapReduce.writable.GroupGenderWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class GroupGenderCountApp {

    public static class GGCMapper extends Mapper<LongWritable, Text, GroupGenderWritable,Text>{

        GroupGenderWritable outKey = new GroupGenderWritable();
        Text outValue = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, GroupGenderWritable, Text>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            outKey.setGroupName(fields[0]);
            outKey.setUserName(fields[1]);
            outKey.setAge(Integer.parseInt(fields[2]));

            outValue.set(fields[3]);

            context.write(outKey,outValue);

        }
    }

    public static class GGCReducer extends Reducer<GroupGenderWritable,Text,Text, NullWritable>{

        Text outKey = new Text();

        @Override
        protected void reduce(GroupGenderWritable key, Iterable<Text> values, Reducer<GroupGenderWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {

            int boySum = 0;
            int girlSum = 0;

            for (Text value : values) {
                String gender = value.toString();
                if(gender.equals("男")){
                    boySum++;
                } else if (gender.equals("女")) {
                    girlSum++;
                }
            }

            String line = key.getGroupName()+"\t"+ key.getAge()+"\t"+"男:"+boySum+";女:"+girlSum;
            outKey.set(line);
            context.write(outKey,NullWritable.get());
        }
    }

    public static void main(String[] args) {

        JobSubmit.submitBaseJob(
                GroupGenderCountApp.class,
                args,
                "group",
                GroupGenderCountGroupingComparator.class);

    }

}
